package com.garden.rocketmq.learn.demo.producer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 批量生产者 发送批量消息
 * 用于较小消息的发送场景
 *
 * 发送条件：
 * 1.批量消息相同topic
 * 2.相同waitStoreMsgOK
 * 3.非延时消息
 * 4.每批发送的大小不超过4m
 * Created by Garden on 2020-08-19 21:54
 */
public class BatchProducer {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException, MQBrokerException {
        //实例化消息生产者
        DefaultMQProducer producer = new DefaultMQProducer("batch_producer_group");
        //设置NameServer的地址，Producer启动后通过NameServer获取目标Topic的路由信息
        producer.setNamesrvAddr("localhost:9876");
        //启动Producer实例
        producer.start();

        //发送消息
        //已知批量消息大小不超过4m
        List<Message> lessThan4mMsgList = new ArrayList<>();
        lessThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.1"
                ,"BatchProducer Msg.1".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        lessThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.2"
                ,"BatchProducer Msg.2".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        lessThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.3"
                ,"BatchProducer Msg.3".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        producer.send(lessThan4mMsgList);
        //未知批量消息大小是否超过4m
        List<Message> greaterThan4mMsgList = new ArrayList<>();
        greaterThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.4"
                ,"BatchProducer Msg.4".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        greaterThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.5"
                ,"BatchProducer Msg.5".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        greaterThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.6"
                ,"BatchProducer Msg.6".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        greaterThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.7"
                ,"BatchProducer Msg.7".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        greaterThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.8"
                ,"BatchProducer Msg.8".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        greaterThan4mMsgList.add(new Message("test_topic","batch_producer_tag","batch_producer_key.9"
                ,"BatchProducer Msg.9".getBytes(RemotingHelper.DEFAULT_CHARSET)));
        //自定义消息分割器，使用消息分割器发送批量消息，保证批量消息大小不超过4m
        ListSplitter listSplitter = new ListSplitter(greaterThan4mMsgList);
        while (listSplitter.hasNext()) {
            List<Message> messages = listSplitter.next();
            producer.send(messages);
        }
    }

    /**
     * 消息分割器
     */
    private static class ListSplitter implements Iterator<List<Message>> {
        //批量消息大小最大值
        private final int SIZE_LIMIT = 1024 * 1024 * 4;
        //待发送的批量消息
        private final List<Message> messages;
        //当前消息下标
        private int currIndex;
        //消息分割器构造器，初始化待发送的批量消息以及当前消息下标
        public ListSplitter(List<Message> messages) {
            this.messages = messages;
            this.currIndex = 0;
        }
        //判断批量消息是否存在下一个批量消息
        @Override public boolean hasNext() {
            return currIndex < messages.size();
        }
        //获取下一个批量消息
        @Override public List<Message> next() {
            int startIndex = getStartIndex();
            int nextIndex = startIndex;
            int totalSize = 0;
            for (; nextIndex < messages.size(); nextIndex++) {
                Message message = messages.get(nextIndex);
                int tmpSize = calcMessageSize(message);
                if (tmpSize + totalSize > SIZE_LIMIT) {
                    break;
                } else {
                    totalSize += tmpSize;
                }
            }
            List<Message> subList = messages.subList(startIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }
        //获取批量消息开始下标
        private int getStartIndex() {
            return currIndex;
        }
        //计算消息大小
        private int calcMessageSize(Message message) {
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
            return tmpSize;
        }
    }

}
